package com.example

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.flink.api.common.functions.FlatMapFunction
import org.apache.flink.api.common.restartstrategy.RestartStrategies
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.common.time.Time
import org.apache.flink.api.common.typeinfo.{TypeInformation, Types}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.types.Row
import org.apache.flink.util.Collector

import java.text.SimpleDateFormat
import java.util.{Locale, Properties}


object WikiPvUv {

  case class WikiLog(userId: String, path: String, timeStamp: Long, refer: String, cookie: String)

  def main(args: Array[String]) {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setRestartStrategy(RestartStrategies.failureRateRestart(
      10, // max-failures-per-interval
      Time.minutes(5), // failure-rate-interval
      Time.seconds(15) // delay
    ));

    //    env.enableCheckpointing(TimeUnit.SECONDS.toMillis(60))

    // 创建 TableEnvironment
    val tableEnv = StreamTableEnvironment.create(env)

    val kafkaProperties: Properties = new Properties
    kafkaProperties.setProperty("bootstrap.servers", "10.5.218.11:6667,10.5.218.12:6667,10.5.218.13:6667")
    kafkaProperties.setProperty("group.id", "wiki_test")

    val consumer: FlinkKafkaConsumer011[String] = new FlinkKafkaConsumer011[String]("wiki_prod_ng_log", new SimpleStringSchema, kafkaProperties)
    val conf: FlinkJedisPoolConfig = new FlinkJedisPoolConfig.Builder().setHost("10.41.252.53").setPort(6379).build

    val dataStream = env.addSource(consumer).flatMap(new FlatMapFunction[String, WikiLog] {
      override def flatMap(t: String, collector: Collector[WikiLog]): Unit = {
        var index = 0
        var userId = ""
        var path = ""
        var refer = ""
        var cookie = ""
        var timeStamp = 0L

        val json: JSONObject = JSON.parseObject(t)
        for (field <- json.get("message").toString.split("]")) {
          if (index == 0) {
            //20/Apr/2020:09:46:56 +0800
            val fm = new SimpleDateFormat("dd/MMM/yyyy:hh:mm:ss Z", Locale.ENGLISH)
            val dt = fm.parse(field.substring(1))
            timeStamp = dt.getTime()
          }
          if (index == 6) {
            userId = field.substring(1)
          }
          if (index == 8) {
            path = field.substring(1)
          }
          if (index == 10) {
            refer = field.substring(1)
          }
          if (index == 11) {
            cookie = field.substring(1)
          }
          index = index + 1
        }
        collector.collect(new WikiLog(userId, path, timeStamp, refer, cookie))
      }
    })

    //    dataStream.filter(new FilterFunction[WikiLog] {
    //      override def filter(t: WikiLog): Boolean = t.path.contains("PUT") && (t.path.contains("/rest/api/content") || t.path.contains("viewpage.action"))
    //    })
    //      .addSink(new RedisSink[WikiLog](conf, new RedisMapper[WikiLog]() {
    //        override def getCommandDescription: RedisCommandDescription = new RedisCommandDescription(RedisCommand.SADD)
    //
    //        override def getKeyFromData(t: WikiLog): String = "PUT:" + DateTime.now().toString("yyyyMMdd")
    //
    //        override def getValueFromData(t: WikiLog): String = t.userId
    //      })).name("redis-PUT-sink");
    //
    //    dataStream.filter(new FilterFunction[WikiLog] {
    //      override def filter(t: WikiLog): Boolean = t.path.contains("/rest/api/content") || t.path.contains("action")
    //    })
    //      .addSink(new RedisSink[WikiLog](conf, new RedisMapper[WikiLog]() {
    //        override def getCommandDescription: RedisCommandDescription = new RedisCommandDescription(RedisCommand.SADD)
    //
    //        override def getKeyFromData(t: WikiLog): String = "UV:" + DateTime.now().toString("yyyyMMdd")
    //
    //        override def getValueFromData(t: WikiLog): String = t.userId
    //      })).name("redis-uv-sink");


    // 注册 Table
    tableEnv.registerDataStream("table1", dataStream)
    // Table API query
//    val tapiResult = tableEnv.scan("table1").select("count(userId),path.substring(0,4)")
    implicit val tpe: TypeInformation[Row] = Types.ROW(Types.STRING, Types.LONG)
//    tapiResult.toRetractStream[Row].print()

    // SQL query
    val sqlResult  = tableEnv.sqlQuery("SELECT userId,count(userId) FROM table1 group by userId")
//    sqlResult.toRetractStream[Row].print()
    env.execute("wiki process")
  }

}
